﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace MiniCAP.RabbitMQ
{
    public class RabbitMqTransport : IMessageTransport
    {
        private readonly IConnection _connection;
        private readonly string _exchangeName;

        public RabbitMqTransport(string connectionString, string exchangeName = "minicap.default")
        {
            var factory = new ConnectionFactory { Uri = new Uri(connectionString) };
            _connection = factory.CreateConnection();
            _exchangeName = exchangeName;

            // 声明Exchange  
            using var channel = _connection.CreateModel();
            channel.ExchangeDeclare(_exchangeName, ExchangeType.Topic, durable: true);
        }

        public async Task SendAsync(string topic, byte[] message, IDictionary<string, string> headers)
        {
            using var channel = _connection.CreateModel();

            var properties = channel.CreateBasicProperties();
            properties.Persistent = true;
            properties.Headers = headers.ToDictionary(x => x.Key, x => (object)x.Value);

            channel.BasicPublish(_exchangeName, topic, properties, message);
        }

        public async Task SubscribeAsync(string topic, string group, Func<byte[], IDictionary<string, string>, Task> handler)
        {
            var channel = _connection.CreateModel();
            var queueName = $"{group}.{topic}";

            channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);
            channel.QueueBind(queueName, _exchangeName, topic);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += async (model, ea) =>
            {
                var headers = ea.BasicProperties.Headers?.ToDictionary(
                    x => x.Key,
                    x => Encoding.UTF8.GetString((byte[])x.Value)) ?? new Dictionary<string, string>();

                try
                {
                    await handler(ea.Body.ToArray(), headers);
                    channel.BasicAck(ea.DeliveryTag, false);
                }
                catch
                {
                    channel.BasicNack(ea.DeliveryTag, false, true);
                }
            };

            channel.BasicConsume(queueName, false, consumer);
        }
    }
}
